package org.onion.pattern.event.springboot.rocketmq.consume.interceptor;

import lombok.Getter;
import org.onion.pattern.event.consumer.EventConsumeHandler;
import org.onion.pattern.event.consumer.ExternalEventInfo;
import org.onion.pattern.event.consumer.log.EventConsumeLog;
import org.onion.pattern.event.consumer.log.EventConsumeLogInterceptor;
import org.onion.pattern.event.consumer.log.EventConsumeLogStorage;
import org.onion.pattern.event.springboot.rocketmq.config.RocketMQ;
import org.onion.pattern.event.springboot.rocketmq.config.RocketMessageMetadata;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Optional;


public class RocketMQEventConsumeLogInterceptor extends EventConsumeLogInterceptor {

    @Autowired
    @Getter
    private EventConsumeLogStorage logStorage;

    @Override
    protected Optional<EventConsumeLog> toEventConsumeLog(ExternalEventInfo eventInfo, Object metadata_, EventConsumeHandler handler) {
        if (!(metadata_ instanceof RocketMessageMetadata)) {
            return Optional.empty();
        }

        RocketMessageMetadata metadata = (RocketMessageMetadata) metadata_;

        EventConsumeLog consumeLog = new EventConsumeLog();

        consumeLog.setConsumerGroup(metadata.getConsumerGroup());
        consumeLog.setTopic(metadata.getTopic());
        consumeLog.setSubTopic(metadata.getSubTopic());
        consumeLog.setReconsumeTimes(metadata.getReconsumeTimes());
        consumeLog.setPublishedTime(metadata.getStoreTime());
        consumeLog.setMsgId(metadata.getMessageId());
        consumeLog.setChannel(RocketMQ.CHANNEL_NAME);

        return Optional.of(consumeLog);
    }


}
